package com.sh.learning.demo.kafka.consumer.service;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.sh.learning.demo.common.message.RegistrationMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

/**
 * @author songHai
 * @Classname TopicMessageListener
 * @Description 接收信息
 * @Date 2025/4/30 12:00
 */
@Component
public class TopicMessageListener {
    @Autowired
    ObjectMapper objectMapper;

    @KafkaListener(topics = "topic_registration", groupId = "group1")
    public void onRegistrationMessage(@Payload String message, @Header("type") String type) throws Exception {
        RegistrationMessage msg = objectMapper.readValue(message, getType(type));
        System.out.printf("接收注册信息received registration message: %s%n", msg);
    }

    @SuppressWarnings("unchecked")
    private static <T> Class<T> getType(String type) {
        // TODO: use cache:
        try {
            return (Class<T>) Class.forName(type);
        } catch (ClassNotFoundException e) {
            throw new RuntimeException(e);
        }
    }
}
